/**
 * 
 */
package tau.adx.simulator;

import gnu.trove.map.hash.TObjectIntHashMap;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;

import lombok.Data;
import lombok.EqualsAndHashCode;

import org.apache.log4j.Logger;

import tau.adx.common.messages.AdxMessage;
import tau.adx.common.messages.auction.AuctionCanceled;
import tau.adx.common.messages.auction.AuctionMessage;
import tau.adx.common.messages.auction.CancellationReason;
import tau.adx.common.messages.users.UserState;
import tau.adx.common.messages.users.UserStateChangedMessage;
import tau.adx.common.publishers.AdxPublisher;
import tau.adx.common.users.AdxUser;

import com.google.common.eventbus.EventBus;
import com.google.common.eventbus.Subscribe;

/**
 * Message counter. Counts the number of messages sent in the system.
 * 
 * @author greenwald
 * 
 */
@Data
@EqualsAndHashCode(callSuper = false)
public class MessageCounter extends Thread {

	/**
	 * Time to wait between message queue scans, after the queue was depeleted
	 * of messages.
	 */
	private static final long WAITING_TIME = 1000;

	/**
	 * Mapping between a {@link CancellationReason} to the number of
	 * {@link AuctionMessage} messages related to it.
	 */
	private TObjectIntHashMap<CancellationReason> cancellationReasonMessageMap = new TObjectIntHashMap<>();

	/**
	 * {@link Logger} instance.
	 */
	private Logger logger = Logger.getLogger(this.getClass());

	/**
	 * {@link Queue} of {@link AdxMessage} to be parsed.
	 */
	private Queue<AdxMessage> messageQeueue = new LinkedBlockingQueue<>();

	/**
	 * Mapping between a {@link AdxPublisher publisher} to the number of
	 * {@link AuctionMessage} messages related to it.
	 */
	private TObjectIntHashMap<AdxPublisher> publisherToAuctionMessageMap = new TObjectIntHashMap<>();

	/**
	 * Mapping between {@link AdxPublisher publishers} and the times each
	 * {@link AdxUser user} (referring to major attributes only) interacted with
	 * it.
	 */
	private Map<AdxPublisher, TObjectIntHashMap<AdxUser>> publisherToUserMap = new HashMap<>();

	/**
	 * Mapping between a {@link UserState} and a {@link Set} of {@link AdxUser
	 * users} that were in that state.
	 */
	private Map<UserState, Set<AdxUser>> stateToUserMap = new HashMap<>();

	/**
	 * Indicates whether the thread should be stopped.
	 */
	private boolean stop = false;

	/**
	 * Indicates whether the thread should be stopped upon
	 * {@link #messageQeueue} depletion.
	 */
	private boolean stopWhenEmpty = false;

	/**
	 * Counter for the number of {@link UserStateChangedMessage} sent in the
	 * system.
	 */
	private AtomicLong userStateMessages = new AtomicLong();

	/**
	 * Mapping between a {@link AdxUser user} to the number of
	 * {@link AuctionMessage} messages related to it.
	 */
	private TObjectIntHashMap<AdxUser> userToAuctionMessageMap = new TObjectIntHashMap<>();

	/**
	 * Mapping between a {@link AdxUser user} and a {@link Set} of
	 * {@link UserStateChangedMessage} messages related to it.
	 */
	private Map<AdxUser, Set<UserStateChangedMessage>> userToStateChangedMap = new HashMap<>();

	/**
	 * SUbscription method to be called by the {@link EventBus}.
	 * 
	 * @param message
	 */
	@Subscribe
	public void notifyMessage(AdxMessage message) {
		messageQeueue.add(message);
	}

	/**
	 * Identifies a given {@link AdxMessage} and calls the appropriate parsing
	 * methods.
	 * 
	 * @param message
	 *            {@link AdxMessage} to be identified.
	 */
	private void parseMessage(AdxMessage message) {
		if (message instanceof UserStateChangedMessage) {
			updateMessageToUserMap((UserStateChangedMessage) message);
			updateUserToStateChangedMap((UserStateChangedMessage) message);
		} else if (message instanceof AuctionMessage) {
			updateUserToAuctionMap((AuctionMessage) message);
			updatePublisherToAuctionMap((AuctionMessage) message);
			updatePublisherToUserMap((AuctionMessage) message);
			if (message instanceof AuctionCanceled) {
				udpateAuctionCanceledCounter((AuctionCanceled) message);
			}
		}
	}

	/**
	 * @see java.lang.Thread#run()
	 */
	@Override
	public void run() {
		logger.debug("started run");
		while (!stop && (!stopWhenEmpty || messageQeueue.size() != 0)) {
			while (messageQeueue.size() != 0) {
				AdxMessage message = messageQeueue.remove();
				parseMessage(message);
			}
			try {
				Thread.sleep(WAITING_TIME);
			} catch (InterruptedException e) {
				logger.error(
						"Thread was interrupted white sleeping between scans",
						e);
			}
		}
		logger.debug("finished run");
	}

	/**
	 * Updates the {@link #cancellationReasonMessageMap} upon arrival of an
	 * {@link AuctionCanceled} message.
	 * 
	 * @param message
	 */
	private void udpateAuctionCanceledCounter(AuctionCanceled message) {
		cancellationReasonMessageMap.adjustOrPutValue(
				message.getCancellationReason(), 1, 1);
	}

	/**
	 * Updates the {@link #stateToUserMap} upon arrival of a new
	 * {@link UserStateChangedMessage}.
	 * 
	 * @param message
	 *            An {@link UserStateChangedMessage} to be parsed.
	 */
	public void updateMessageToUserMap(UserStateChangedMessage message) {
		if (!stateToUserMap.containsKey(message.getState())) {
			stateToUserMap.put(message.getState(), new HashSet<AdxUser>());
		}
		stateToUserMap.get(message.getState()).add(message.getUser());
	}

	/**
	 * Updates the {@link #publisherToAuctionMessageMap} upon arrival of a new
	 * {@link AuctionMessage}.
	 * 
	 * @param message
	 *            An {@link AuctionMessage} to be parsed.
	 */
	public void updatePublisherToAuctionMap(AuctionMessage message) {
		publisherToAuctionMessageMap.adjustOrPutValue(message.getPublisher(),
				1, 1);
	}

	/**
	 * Updates {@link Publisher} to {@link AdxUser} mapping and counters.
	 * @param message {@link AuctionMessage}.
	 */
	private void updatePublisherToUserMap(AuctionMessage message) {
		if (!publisherToUserMap.containsKey(message.getPublisher())) {
			publisherToUserMap.put(message.getPublisher(),
					new TObjectIntHashMap<AdxUser>());
		}
		TObjectIntHashMap<AdxUser> userMap = publisherToUserMap.get(message
				.getPublisher());

		AdxUser user = (AdxUser) message.getUser().clone();
		user.ignoreMinorAttributes();
		userMap.adjustOrPutValue(user, 1, 1);
	}

	/**
	 * Updates the {@link #userToAuctionMessageMap} upon arrival of a new
	 * {@link AuctionMessage}.
	 * 
	 * @param message
	 *            An {@link AuctionMessage} to be parsed.
	 */
	public void updateUserToAuctionMap(AuctionMessage message) {
		userToAuctionMessageMap.adjustOrPutValue(message.getUser(), 1, 1);
	}

	/**
	 * Updates the {@link #userToStateChangedMap} upon arrival of a new
	 * {@link UserStateChangedMessage}.
	 * 
	 * @param message
	 *            An {@link UserStateChangedMessage} to be parsed.
	 */
	public void updateUserToStateChangedMap(UserStateChangedMessage message) {
		if (!userToStateChangedMap.containsKey(message.getUser())) {
			userToStateChangedMap.put(message.getUser(),
					new HashSet<UserStateChangedMessage>());
		}
		userToStateChangedMap.get(message.getUser()).add(message);
	}
}
